package com.atguigu.edu.app.dws;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.app.BaseApp;
import com.atguigu.edu.bean.TrafficPageViewBean;
import com.atguigu.edu.common.Constant;
import com.atguigu.edu.util.AtguiguUtil;
import com.atguigu.edu.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

public class DwsTrafficVcChArIsNewPageViewWindow extends BaseApp {
    public static void main(String[] args) {
        new DwsTrafficVcChArIsNewPageViewWindow().init(4002,2,
                "DwsTrafficVcChArIsNewPageViewWindow",
                Constant.TOPIC_DWD_TRAFFIC_PAGE);
    }
    @Override
    public void handle(StreamExecutionEnvironment ev, DataStreamSource<String> stream) {
        //解析成pojo类型
        SingleOutputStreamOperator<TrafficPageViewBean> poJoStream = parseToPojo(stream);
        //开窗聚合
        SingleOutputStreamOperator<TrafficPageViewBean> resultStream = windowAndAgg(poJoStream);
        //储存到clickhouse
        writeToClickHouse(resultStream);

    }

    private void writeToClickHouse(SingleOutputStreamOperator<TrafficPageViewBean> resultStream) {
        resultStream.addSink(FlinkSinkUtil.getClickHouseSink("dws_traffic_vc_ch_ar_is_new_page_view_window",
                TrafficPageViewBean.class));
    }

    private SingleOutputStreamOperator<TrafficPageViewBean> windowAndAgg(SingleOutputStreamOperator<TrafficPageViewBean> poJoStream) {
        return poJoStream.assignTimestampsAndWatermarks(WatermarkStrategy
                .<TrafficPageViewBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                .withTimestampAssigner((bean,ts)-> bean.getTs())
                .withIdleness(Duration.ofSeconds(60))
        ).keyBy(bean-> bean.getVc()+"_"+ bean.getCh() + "_" + bean.getAr() + "_" + bean.getIsNew())
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .reduce(new ReduceFunction<TrafficPageViewBean>() {
                    @Override
                    public TrafficPageViewBean reduce(TrafficPageViewBean value1, TrafficPageViewBean value2) throws Exception {
                        value1.setPvCt(value1.getPvCt() + value2.getPvCt());
                        value1.setSvCt(value1.getSvCt() + value2.getSvCt());
                        value1.setUvCt(value1.getUvCt() + value2.getUvCt());
                        value1.setDurSum(value1.getDurSum() + value2.getDurSum());
                        return value1;
                    }
                }, new ProcessWindowFunction<TrafficPageViewBean, TrafficPageViewBean, String, TimeWindow>() {
                    @Override
                    public void process(String s, Context context,
                                        Iterable<TrafficPageViewBean> iterable, Collector<TrafficPageViewBean> collector) throws Exception {
                        TrafficPageViewBean next = iterable.iterator().next();
                        next.setStt(AtguiguUtil.tsToDateTime(context.window().getStart()));
                        next.setEdt(AtguiguUtil.tsToDateTime(context.window().getEnd()));
                        //改ts为结果统计时间 用系统时间
                        next.setTs(System.currentTimeMillis());
                        collector.collect(next);
                    }
                });
    }

    private SingleOutputStreamOperator<TrafficPageViewBean> parseToPojo(DataStreamSource<String> stream) {
        return stream.map(JSONObject::parseObject)
                .keyBy(k -> k.getString("mid"))
                .process(new KeyedProcessFunction<String, JSONObject, TrafficPageViewBean>() {

                    private ValueState<String> lastVisitDate;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        lastVisitDate = getRuntimeContext().getState(new ValueStateDescriptor<String>("lastVisitDate", String.class));
                    }

                    @Override
                    public void processElement(JSONObject obj,
                                               Context context,
                                               Collector<TrafficPageViewBean> collector) throws Exception {
                        String vc = obj.getString("vc");
                        String ch = obj.getString("ch");
                        String ar = obj.getString("ar");
                        String isNew = obj.getString("is_new");
                        Long durSum = obj.getLong("during_time");
                        Long ts = obj.getLong("ts");
                        //获取当天第一条日志记录
                        Long uvCt = 0l;
                        String today = AtguiguUtil.tsToDate(ts);
                        if (!today.equals(lastVisitDate.value())){
                            uvCt=1l;
                            lastVisitDate.update(today);
                        }
                        //获取会话次数
                        Long svCt =0l;
                        if (obj.getString("last_page_id")==null){
                            svCt = 1l;
                        }
                        collector.collect(new TrafficPageViewBean(
                                "", "",
                                vc, ch, ar, isNew,
                                uvCt, svCt, 1L, durSum,
                                ts));
                    }
                });
    }
}